package com.plumelog.server.collect;

import com.alibaba.fastjson.JSON;
import com.plumelog.core.client.AbstractClient;
import com.plumelog.core.constant.LogMessageConstant;
import com.plumelog.core.dto.RunLogMessage;
import com.plumelog.server.InitConfig;
import com.plumelog.server.client.ElasticLowerClient;
import com.plumelog.server.util.IndexUtil;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationEventPublisher;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.*;

/**
 * className：KafkaLogCollect
 * description：KafkaLogCollect 获取kafka中日志，存储到es
 *
 * @author Frank.chen
 * @version 1.0.0
 */
public class KafkaLogCollect extends BaseLogCollect {
    private final org.slf4j.Logger logger = LoggerFactory.getLogger(KafkaLogCollect.class);
    private AbstractClient redisClient;
    private KafkaConsumer<String, String> kafkaConsumer;

    public KafkaLogCollect(ElasticLowerClient elasticLowerClient, KafkaConsumer kafkaConsumer, ApplicationEventPublisher applicationEventPublisher, AbstractClient redisClient) {
        super.elasticLowerClient = elasticLowerClient;
        this.kafkaConsumer = kafkaConsumer;
        this.kafkaConsumer.subscribe(Arrays.asList(LogMessageConstant.LOG_KEY, LogMessageConstant.LOG_KEY_TRACE));
        super.applicationEventPublisher = applicationEventPublisher;
        super.redisClient = redisClient;
        logger.info("kafkaConsumer subscribe ready!");
        logger.info("sending log ready!");
    }

    public void kafkaStart() {
        threadPoolExecutor.execute(() -> {
            collectRuningLog();
        });
        logger.info("KafkaLogCollect is starting!");
    }

    public void collectRuningLog() {
        while (true) {
            List<String> logList = new ArrayList();
            List<String> sendlogList = new ArrayList();
            try {
                ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofMillis(1000));
                records.forEach(record -> {
                    if (logger.isDebugEnabled()) {
                        logger.debug("get log:" + record.value() + "  logType:" + record.topic());
                    }
                    if (record.topic().equals(LogMessageConstant.LOG_KEY)) {
                        logList.add(record.value());
                    }
                    if (record.topic().equals(LogMessageConstant.LOG_KEY_TRACE)) {
                        sendlogList.add(record.value());
                    }
                });
            } catch (Exception e) {
                logger.error("get logs from kafka failed! ", e);
            }
            if (logList.size() > 0) {
                if ("his".equals(InitConfig.LOG_SAVEINDEX_TYPE)) {
                    Map<String, List<String>> procMapList = new HashMap<>();
                    for (int i = 0; i < logList.size(); i++) {
                        String log = logList.get(i);
                        RunLogMessage runLogMessage = JSON.parseObject(log, RunLogMessage.class);
                        String dateTime = runLogMessage.getDateTime();

                        SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
                        Date date = new Date();
                        try {
                            date = sf.parse(dateTime);
                        } catch (ParseException e) {
                            e.printStackTrace();
                        }

                        String baseIndex = "";
                        if ("day".equals(InitConfig.ES_INDEX_MODEL)) {
                            baseIndex = IndexUtil.getRunLogIndex(date.getTime());
                        } else {
                            baseIndex = IndexUtil.getRunLogIndexWithHour(date.getTime());
                        }
                        List<String> stringList = procMapList.getOrDefault(baseIndex, new ArrayList());
                        stringList.add(log);
                        procMapList.put(baseIndex, stringList);
                    }
                    procMapList.forEach((key, value) -> {
                        super.sendLog(key, value);
                    });
                } else {
                    super.sendLog(super.getRunLogIndex(), logList);
                }

                publisherMonitorEvent(logList);
            }
            if (sendlogList.size() > 0) {
                super.sendTraceLogList(super.getTraceLogIndex(), sendlogList);
            }
        }
    }
}
